﻿using AliIotAmqpWorker.Common;
using AliIotLib;
using Amqp;
using Amqp.Framing;
using Amqp.Sasl;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Net.Security;
using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Threading;

namespace AliIotAmqpWorker
{
    public class AliAmqp
    {
        public string Host = "1701930854581949.iot-amqp.cn-shanghai.aliyuncs.com";//
        public int Port = 5671;
        public int Count = 0;
        public int IntervalTime = 10000;
        public Address address;
 
        private readonly MongoDBHelper _mongoDBHelper;
        private readonly AliIotConfig _aliIotConfig;

        public AliAmqp(MongoDBHelper mongoDBHelper, AliIotConfig aliIotConfig)
        {
            _mongoDBHelper = mongoDBHelper;
            _aliIotConfig = aliIotConfig;
            Host = $"{_aliIotConfig.UserId}.iot-amqp.{_aliIotConfig.RegionId}.aliyuncs.com";
        }
        /// <summary>
        /// 初始化
        /// </summary>
        public void Initial()
        {
            long timestamp = GetCurrentMilliseconds();
            string param = "authId=" + _aliIotConfig.AccessKeyId + "&timestamp=" + timestamp;
            //按照阿里云IoT规范，组装UserName。
            string username = _aliIotConfig.ClientId + "|authMode=aksign,signMethod=hmacmd5,consumerGroupId=" + _aliIotConfig.ConsumerGroupId
               + ",iotInstanceId=" + _aliIotConfig.IotInstanceId + ",authId=" + _aliIotConfig.AccessKeyId + ",timestamp=" + timestamp + "|";
            //按照阿里云IoT规范，计算签名，组装password
            string password = doSign(param, _aliIotConfig.AccessKeySecret, "HmacMD5");

            DoConnectAmqp(username, password);

            //ManualResetEvent resetEvent = new ManualResetEvent(false);
            //resetEvent.WaitOne();
        }

        public void DoConnectAmqp(string username, string password)
        {
            address = new Address(Host, Port, username, password);
            //Create Connection
            ConnectionFactory cf = new ConnectionFactory();
            //use local tls if neccessary
            //cf.SSL.ClientCertificates.Add(GetCert());
            //cf.SSL.RemoteCertificateValidationCallback = ValidateServerCertificate;
            cf.SASL.Profile = SaslProfile.External;
            cf.AMQP.IdleTimeout = 120000;
            cf.AMQP.ContainerId = "client.1.2";
            cf.AMQP.HostName = "contoso.com";
            cf.AMQP.MaxFrameSize = 8 * 1024;
            var connection = cf.CreateAsync(address).Result;

            //Connection Exception Closed
            connection.AddClosedCallback(ConnClosed);

            //Receive Message
            DoReceive(connection);
        }
        /// <summary>
        /// 接收
        /// </summary>
        /// <param name="connection"></param>
        public void DoReceive(Connection connection)
        {
            //Create Session
            var session = new Session(connection);

            //Create Link and Receive Message
            var receiver = new ReceiverLink(session, "queueName", null);

            receiver.Start(20, (link, message) =>
            {
                object messageId = message.ApplicationProperties["messageId"];
                object topic = message.ApplicationProperties["topic"];
                string body = Encoding.UTF8.GetString((Byte[])message.Body);
                //注意：此处不要有耗时的逻辑，如果这里要进行业务处理，请另开线程，否则会堵塞消费。如果消费一直延时，会增加消息重发的概率。
                Console.WriteLine(DateTime.Now.ToLongTimeString());
                Console.WriteLine("receive message, topic=" + topic);
                Console.WriteLine("messageId=" + messageId);
                Console.WriteLine("body=" + body);
                try
                {
                    MongoDbTag mongoDbTag = new MongoDbTag();
                    if (body.Contains("status"))
                    {
                        AliIotDeviceStatus status = JsonConvert.DeserializeObject<AliIotDeviceStatus>(body);
                        if (status != null)
                        {
                            //mongoDbTag.device_name = status.deviceName;
                            //mongoDbTag.value = body;
                            //_mongoDBHelper.InsertOne<MongoDbTag>(mongoDbTag, "raw_status");
                        }
                    }
                    else
                    {
                        AliIotDeviceData<ECU1151Info> ecu1151 = JsonConvert.DeserializeObject<AliIotDeviceData<ECU1151Info>>(body);
                        if (ecu1151 != null)
                        {
                            mongoDbTag.device_name = ecu1151.deviceName;
                            mongoDbTag.value = body;
                            _mongoDBHelper.InsertOne<MongoDbTag>(mongoDbTag, "raw_data");
                        }
                    }
                }
                catch(Exception ex)
                {
                    Console.WriteLine("消息解析出错："+ex.Message);
                }
                
                Console.WriteLine("");
                
                link.Accept(message);//Acknowledge Message
            });
        }
        //连接发生异常后，进入重连模式。
        //这里只是一个简单重试的示例，您可以采用指数退避方式，来完善异常场景，重连策略。
        public void ConnClosed(IAmqpObject _, Error e)
        {
            Console.WriteLine("ocurr error: " + e);
            if (Count < 3)
            {
                Count += 1;
                Thread.Sleep(IntervalTime * Count);
            }
            else
            {
                Thread.Sleep(120000);
            }

            //Reconnection
            DoConnectAmqp(address.User, address.Password);
        }

        public X509Certificate GetCert()
        {
            string certPath = Environment.CurrentDirectory + "/root.crt";
            X509Certificate crt = new X509Certificate(certPath);

            return crt;
        }

        public bool ValidateServerCertificate(object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
        {
            return true;
        }

        public long GetCurrentMilliseconds()
        {
            DateTime dt1970 = new DateTime(1970, 1, 1);
            DateTime current = DateTime.Now;
            return (long)(current - dt1970).TotalMilliseconds;
        }

        //签名方法：支持hmacmd5，hmacsha1和hmacsha256
        public string doSign(string param, string accessSecret, string signMethod)
        {
            //signMethod = HmacMD5
            byte[] key = Encoding.UTF8.GetBytes(accessSecret);
            byte[] signContent = Encoding.UTF8.GetBytes(param);
            var hmac = new HMACMD5(key);
            byte[] hashBytes = hmac.ComputeHash(signContent);
            return Convert.ToBase64String(hashBytes);
        }
    }
}
